use crate::admin_api::{
    g_db_backups, g_db_keys, g_db_value, g_statistics, p_admin_role, p_db_backup, p_db_edit_value,
    p_remove_by_head, p_remove_db_keys,
};
use crate::vdd_prometheus::{
    PrometheusFilter, MODEL_COUNT_GAUGE, REQUEST_COUNTER, UP_GAUGE, USER_COUNT_GAUGE,
};
use chrono::Local;
use jsonwebtoken::EncodingKey;
use salvo::catcher::Catcher;
use salvo::conn::rustls::{Keycert, RustlsConfig};
use salvo::conn::TcpListener;
use salvo::core::prelude::*;
use salvo::jwt_auth::{ConstDecoder, JwtAuth, JwtAuthDepotExt};
use salvo::prelude::StaticDir;
use salvo::{Listener, Server};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::net::SocketAddr;
use std::str::FromStr;
use std::sync::LazyLock;
use std::time::{SystemTime, UNIX_EPOCH};
use tokio_cron::{Job, Scheduler};
use visit_dd_core::id_factory::ID_FACTORY;
use visit_dd_core::proj::APP_CONFIG;
use visit_dd_core::proj::{VddUser, VddUserRole};
use visit_dd_core::{
    i18n, ActionReturnMessage, Ares, Astr, FieldId, RecordId, UserId, SECRET_KEY, TYPE_DICT,
};
use visit_dd_db::{disconnect_db, ROCKSDB};

mod admin_api;
mod api;
mod field_api;
mod gateway_api;
mod model_api;
mod query_api;
mod record_api;
#[cfg(test)]
mod test;
mod vdd_polars;
mod vdd_prometheus;
mod visit_api;

#[derive(Serialize, Deserialize)]
pub struct JwtClaims {
    pub uid: UserId,
    pub role: VddUserRole,
    pub exp: i64,
}
///
impl JwtClaims {
    pub fn check_admin_role(&self) -> Option<Astr> {
        if self.role != VddUserRole::UrAdmin {
            return Some(i18n!(9001));
        }
        None
    }
    const LIVE_SECONDS: i64 = 86400;
    //一个小时后超时
    pub fn from_user(user: &VddUser) -> Self {
        Self {
            uid: user.id.clone(),
            exp: Local::now().timestamp() + Self::LIVE_SECONDS,
            role: user.role.to_owned(),
        }
    }
    pub fn refresh_life(&self) -> Self {
        Self {
            uid: self.uid.to_owned(),
            exp: Local::now().timestamp() + Self::LIVE_SECONDS,
            role: self.role.to_owned(),
        }
    }
}
///
pub trait WebResponseSupportTrait {
    fn error(&mut self, message: &str);
    fn ok_empty(&mut self);
    fn ok<T: Sized + Serialize>(&mut self, data: &T);
    fn wrap(&mut self, r: Ares);
}

impl WebResponseSupportTrait for Response {
    fn error(&mut self, message: &str) {
        self.render(json!(&ActionReturnMessage::error((), message)).to_string());
    }

    fn ok_empty(&mut self) {
        self.render(json!(&ActionReturnMessage::ok(())).to_string());
    }

    fn ok<T: Sized + Serialize>(&mut self, data: &T) {
        self.render(json!(&ActionReturnMessage::ok(data)).to_string());
    }

    fn wrap(&mut self, r: Ares) {
        if r.is_ok() {
            self.ok_empty();
        } else {
            self.error(r.unwrap_err().as_ref());
        }
    }
}

pub struct JwtOutTimeFilter;

impl JwtOutTimeFilter {
    pub fn new() -> Self {
        Self
    }
}

#[async_trait]
impl Handler for JwtOutTimeFilter {
    async fn handle(
        &self,
        req: &mut Request,
        depot: &mut Depot,
        res: &mut Response,
        ctrl: &mut FlowCtrl,
    ) {
        let jwt = depot.jwt_auth_data::<JwtClaims>();
        if jwt.is_none() {
            res.render(StatusError::unauthorized());
            ctrl.skip_rest();
        } else {
            let jwt = &jwt.unwrap().claims;
            if Local::now().timestamp() >= jwt.exp {
                res.render(StatusError::unauthorized());
                ctrl.skip_rest();
            } else {
                let new_jwt = jwt.refresh_life();
                let _ = res.add_header("token", to_token(&new_jwt), true);
                ctrl.call_next(req, depot, res).await;
            }
        }
    }
}

/// 系统启动时间戳
static START_TS: LazyLock<u64> = LazyLock::new(|| {
    let ts = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_secs();
    ts
});

fn to_token(jwt_claims: &JwtClaims) -> String {
    jsonwebtoken::encode(
        &jsonwebtoken::Header::default(),
        jwt_claims,
        &EncodingKey::from_secret(SECRET_KEY.as_bytes()),
    )
    .unwrap()
}

#[macro_export]
macro_rules! p {
    ($path:literal, $func:expr) => {
        Router::with_path($path).post($func)
    };
}

#[macro_export]
macro_rules! g {
    ($path:literal, $func:expr) => {
        Router::with_path($path).get($func)
    };
}

fn create_router() -> Router {
    fn _gateway_router() -> Router {
        Router::with_path("gateway")
            .push(p!("page/{model-tag}", gateway_api::read_record_page))
            .push(p!("batch-add/{model-tag}", gateway_api::append_records))
            .push(p!("batch-modify/{model-tag}", gateway_api::modify_records))
            .push(p!("check-exist/{model-tag}", gateway_api::check_records))
            .push(g!("export-fv/{model-tag}", gateway_api::export_fv))
            .push(p!(
                "export-grouped-fv/{model-tag}",
                gateway_api::export_grouped_fv
            ))
            .push(g!("public-key", api::g_public_key))
    }
    fn _admin_router() -> Router {
        Router::with_path("api-admin")
            .hoop(JwtAuth::<JwtClaims, ConstDecoder>::new(
                ConstDecoder::from_secret(SECRET_KEY.as_bytes()),
            ))
            .hoop(JwtOutTimeFilter::new())
            .push(
                Router::with_path("db")
                    .push(g!("keys", g_db_keys))
                    // .push(Router::with_path("sub-keys",g_db_sub_keys))
                    .push(g!("value", g_db_value))
                    .push(p!("edit-value", p_db_edit_value))
                    .push(p!("remove", p_remove_db_keys))
                    .push(p!("remove-by-head", p_remove_by_head))
                    .push(p!("backup", p_db_backup))
                    .push(g!("list-backups", g_db_backups)),
            )
            .push(g!("statistics", g_statistics))
            .push(p!("admin-role", p_admin_role))
    }
    fn _design_router() -> Router {
        Router::with_path("model")
            .push(p!("save", model_api::p_save_model))
            .push(p!("create", model_api::p_create_model))
            .push(p!("remove", model_api::p_remove))
            .push(p!("export", model_api::p_export))
            .push(p!("download", model_api::p_download_parquet))
            .push(g!("mine", model_api::g_mine))
            .push(p!("log", api::g_model_logs))
            .push(g!("mine-with-tag", field_api::g_as_tree_with_tag))
            .push(g!(
                "list-field/{model-id:num}",
                field_api::g_fields_in_model
            ))
            .push(p!("create-field", field_api::p_create_field))
            .push(p!("change-field", field_api::p_change_field))
            .push(p!("remove-field/{field-id:num}", field_api::p_remove_field))
            .push(p!(
                "re-import-field/{model-id:num}",
                field_api::p_re_import_fields
            ))
            .push(p!(
                "/check-parquet/{file-key:num}",
                model_api::p_handle_parquet
            ))
            .push(p!(
                "/drop-parquet/{file-key:num}",
                model_api::p_drop_parquet
            ))
            .push(p!("/import-file", model_api::p_import))
            .push(p!("/drop-field", field_api::p_drop))
            .push(p!("/drop", model_api::p_drop))
            .push(p!("/restore", model_api::p_restore))
            .push(g!(
                "/list-use-field/{fid:num}",
                model_api::g_models_use_field
            ))
    }
    fn _share_router() -> Router {
        Router::with_path("share")
            .push(g!("vgs/{model-id:num}", visit_api::g_visit_groups))
            .push(p!("save-vg", visit_api::p_save_vg))
            .push(p!("remove-vg", visit_api::p_remove_vg))
            .push(g!("requests/{model-id:num}", visit_api::g_requests))
            .push(p!("change-request", visit_api::p_handle_request))
            .push(p!("check-auth/{model-id:num}", visit_api::p_check_auth))
            .push(g!("all", visit_api::g_shared_models))
            .push(p!("send-request", visit_api::p_send_request))
            .push(p!("verify-code", visit_api::p_verify_code))
            .push(p!("save-tags", visit_api::p_save_tags))
            .push(g!("tags/{model-id:num}", field_api::g_tag_map))
            .push(g!("my-token", visit_api::g_my_token))
            .push(p!("drop-token/{model-id:num}", visit_api::p_drop_token))
            .push(g!("tag-status", visit_api::g_tag_status))
            .push(p!(
                "subscribe-changes/{model-id:num}",
                record_api::p_subscribe_record_change
            ))
            .push(p!(
                "cancel-subscribe/{model-id:num}",
                record_api::p_cancel_subscribe
            ))
            .push(p!(
                "reset-subscription/{model-id:num}",
                record_api::p_reset_subscription
            ))
            .push(g!("sample/{model-id:num}", visit_api::g_record_sample))
    }
    fn _record_router() -> Router {
        Router::with_path("record")
            .push(g!("count-record", record_api::g_count_record))
            .push(p!("save", record_api::p_save_record))
            .push(p!("share-save", record_api::p_share_save))
            .push(p!("page", record_api::p_page))
            .push(p!("share-page", record_api::p_share_page))
            .push(p!("remove", record_api::p_remove_record))
            .push(g!("rvrs/{model-id:num}", field_api::g_rvrs))
            .push(p!("save-rvr", field_api::p_save_rvr))
            .push(p!("remove-rvr", field_api::p_remove_rvr))
            .push(g!(
                "version-count/{record-id:num}",
                record_api::g_version_count
            ))
            .push(g!("versions/{record-id:num}", record_api::g_versions))
            .push(p!("remove-version", record_api::p_remove_version))
            .push(p!("restore-version", record_api::p_restore_version))
            .push(p!("subscription-page", record_api::p_subscription_page))
            .push(g!(
                "subscription-overview/{model-id:num}",
                record_api::g_subscription_overview
            ))
            .push(g!("trash", record_api::g_trash_tree))
            .push(p!("in-trash", record_api::g_records_in_trash))
            .push(p!("drop-from-trash", record_api::p_drop_from_trash))
            .push(p!("restore", record_api::p_restore))
            .push(g!(
                "load/{model-id:num}/{record-id:num}",
                record_api::g_record
            ))
    }
    fn _user_router() -> Router {
        Router::with_path("user")
            .push(p!("check-token", api::p_check_token))
            .push(p!("log", api::g_user_logs))
            .push(g!("message", api::g_messages))
            .push(g!("new-message-count", api::g_new_message_count))
            .push(p!("send-message", api::p_send_message))
            .push(g!("others", api::g_others))
            .push(p!("read-message", api::p_read_message))
            .push(p!("change-pwd", api::p_change_password))
            .push(g!("about-me", api::g_about_me))
            .push(p!("change-name", api::p_change_name))
            .push(p!("change-key", api::p_change_key))
            .push(p!("download-pri-key", api::p_download_pri_key))
    }
    fn _common_router() -> Router {
        Router::with_path("common")
            .push(p!("upload-file", api::upload_file))
            .push(g!("download-file/{file_key:num}", api::download_file))
            .push(g!("codes/{cf}", api::g_codes))
    }
    fn _query_router() -> Router {
        Router::with_path("query")
            .push(g!("my-plan", query_api::g_my_plan))
            .push(p!("save-plan", query_api::p_save_plan))
            .push(p!("remove-plan", query_api::p_remove_plan))
            .push(p!("try-plan", query_api::p_try_plan))
            .push(p!("exec-plan", query_api::p_exec_plan))
            .push(p!("share-query", query_api::p_run_share_sql))
    }
    Router::new()
        .hoop(PrometheusFilter::new())
        .push(Router::new().get(api::index))
        .push(g!(
            "web/{*path}",
            StaticDir::new(["web"])
                .defaults("index.html")
                .fallback("index.html")
                .auto_list(false)
        ))
        .push(_admin_router())
        .push(_gateway_router())
        .push(
            Router::with_path("api/v1")
                .push(g!("prometheus", vdd_prometheus::g_prometheus))
                .push(g!("pub-key", api::g_public_key))
                .push(g!("locale", api::g_locale))
                .push(p!("user/login", api::p_user_log_in))
                .push(p!("user/register", api::p_user_register))
                .push(
                    Router::new()
                        .hoop(JwtAuth::<JwtClaims, ConstDecoder>::new(
                            ConstDecoder::from_secret(SECRET_KEY.as_bytes()),
                        ))
                        .hoop(JwtOutTimeFilter::new())
                        .push(_common_router())
                        .push(_user_router())
                        .push(_design_router())
                        .push(_share_router())
                        .push(_record_router())
                        .push(_query_router()),
                ),
        )
}

fn extract_record_ids(j: &Value) -> Vec<RecordId> {
    let record_ids = j["recordIds"].as_array();
    match record_ids {
        None => Vec::<RecordId>::new(),
        Some(record_ids) => record_ids
            .iter()
            .map(|j2| j2.as_u64().map(|id| RecordId(id)))
            .filter(|id| id.is_some())
            .map(|id| id.unwrap())
            .collect::<Vec<_>>(),
    }
}

fn extract_field_ids(j: &Value) -> Vec<FieldId> {
    let fields = j["fieldIds"].as_array();
    match fields {
        None => Vec::<FieldId>::new(),
        Some(field_ids) => field_ids
            .iter()
            .map(|field| field.as_u64().map(|id| FieldId(id)))
            .filter(|id| id.is_some())
            .map(|id| id.unwrap())
            .collect::<Vec<_>>(),
    }
}

#[tokio::main]
async fn main() {
    let file_appender = tracing_appender::rolling::daily("log/", "visit_dd_log");
    let (non_blocking, _guard) = tracing_appender::non_blocking(file_appender);
    tracing_subscriber::fmt()
        .pretty()
        .with_writer(non_blocking)
        .with_file(false)
        .with_ansi(false)
        .init();

    let config = LazyLock::force(&APP_CONFIG);
    LazyLock::force(&TYPE_DICT);
    LazyLock::force(&ROCKSDB);
    LazyLock::force(&ID_FACTORY);
    LazyLock::force(&UP_GAUGE);
    LazyLock::force(&USER_COUNT_GAUGE);
    LazyLock::force(&MODEL_COUNT_GAUGE);
    LazyLock::force(&REQUEST_COUNTER);

    let mut scheduler = Scheduler::local();
    scheduler.add(Job::new(
        "* */30 * * * *",
        visit_dd_db::get_rocks_statistics,
    ));

    let router = create_router();
    let service = Service::new(router).catcher(Catcher::default());

    let address = if config.ipv6 {
        SocketAddr::from_str(&format!("[::]:{}", config.server_port)).unwrap()
    } else {
        SocketAddr::from_str(&format!("0.0.0.0:{}", config.server_port)).unwrap()
    };

    let ssl_pair = RustlsConfig::new(
        Keycert::new()
            .cert(include_bytes!("visitdd.crt").as_ref())
            .key(include_bytes!("pri_1.key").as_ref()),
    );
    let listener = TcpListener::new(address).rustls(ssl_pair).bind().await;
    let server = Server::new(listener);
    LazyLock::force(&START_TS);
    server.serve(service).await;
    disconnect_db();
}
